package mq

import (
	"animal/octopus/common"
	"animal/octopus/config"
	"animal/octopus/core"
	"encoding/json"
	"sync"

	"github.com/streadway/amqp"

	log "github.com/sirupsen/logrus"
)

// ------------------------------------------------------------------------------
// singleton
// ------------------------------------------------------------------------------
type MQServer struct {
	Conn  *amqp.Connection
	Chan  *amqp.Channel
	Queue amqp.Queue
}

var instance *MQServer
var once sync.Once

func Instance() *MQServer {
	once.Do(func() {
		instance = &MQServer{}
	})
	return instance
}

// ------------------------------------------------------------------------------
//
// ------------------------------------------------------------------------------
func (server *MQServer) Init() bool {
	if config.Instance().Data.RabbitMQ.URL == "" {
		log.Info("no rabbitmq configure")
		return true
	}

	if config.Instance().Data.RabbitMQ.Exchange == "" {
		log.Error("cannot find rabbitmq exchange")
		return false
	}

	var err error
	var result bool = false
	var exchangeName string = config.Instance().Data.RabbitMQ.Exchange

	defer func() {
		if !result {
			if server.Chan != nil {
				server.Chan.Close()
				server.Chan = nil
			}

			if server.Conn != nil {
				server.Conn.Close()
				server.Conn = nil
			}
		}
	}()

	// 连接RabbitMQ服务器
	server.Conn, err = amqp.Dial(config.Instance().Data.RabbitMQ.URL)
	if err != nil {
		log.WithFields(log.Fields{
			"url":   config.Instance().Data.RabbitMQ.URL,
			"error": err,
		}).Error("cannot connect to rabbitmq server")
		return false
	}

	// 创建一个通道
	server.Chan, err = server.Conn.Channel()
	if err != nil {
		log.WithField("error", err).Error("cannot open channel")
		return false
	}

	// 声明exchange，如果不存在则创建
	err = server.Chan.ExchangeDeclare(
		exchangeName,
		"fanout",
		false, // 不持久化
		false, // 不自动删除
		false, // 不内部使用
		false, // 不等待确认
		nil,
	)
	if err != nil {
		log.WithFields(log.Fields{"exchange": exchangeName, "error": err}).
			Error("cannot declare exchange")
		return false
	}

	// 创建一个匿名队列
	server.Queue, err = server.Chan.QueueDeclare(
		"",    // 队列名称为空，由RabbitMQ自动生成
		false, // 不持久化
		false, // 不自动删除
		true,  // 排他性队列，只对当前连接可见
		false, // 不等待确认
		nil,
	)
	if err != nil {
		log.WithField("error", err).Error("cannot declare queue")
		return false
	}

	// 绑定队列到exchange上
	err = server.Chan.QueueBind(
		server.Queue.Name,
		"",           // 路由键为空，表示接收所有消息
		exchangeName, // 绑定到指定的exchange上
		false,
		nil,
	)
	if err != nil {
		log.WithFields(log.Fields{
			"queue":    server.Queue.Name,
			"exchange": exchangeName,
			"error":    err,
		}).Fatalf("cannot bind queue to exchange")
		return false
	}

	result = true

	log.Info("init rabbitmq connection success")

	return true
}

// ------------------------------------------------------------------------------
//
// ------------------------------------------------------------------------------
func (server *MQServer) Run() {
	if server.Chan == nil {
		return
	}

	go func() {
		msgs, err := server.Chan.Consume(
			server.Queue.Name,
			"",
			true,  // 自动确认消息
			false, // 不使用独立的消费者标识
			false, // 不等待确认
			false, // 不限制消费者个数
			nil,
		)
		if err != nil {
			log.WithField("error", err).Error("cannot consume message")
			return
		}

		for msg := range msgs {
			server.processMessage(string(msg.Body))
		}
	}()
}

// ------------------------------------------------------------------------------
//
// ------------------------------------------------------------------------------
func (server *MQServer) processMessage(msg string) {
	log.WithField("msg", msg).Info("received mq message")

	var report common.ReportInfo
	err := json.Unmarshal([]byte(msg), &report)
	if err != nil {
		log.WithFields(log.Fields{
			"msg":   msg,
			"error": err,
		}).Error("unmarshal report failed")
	}

	if core.Report(nil, report) != nil {
		log.Error("process report failed")
	}
}
